package com.ruoyi.ai.factory.impl;

import com.alibaba.dashscope.common.Role;
import com.google.common.collect.Lists;
import com.ruoyi.ai.domain.bo.WxDialogueBo;
import com.ruoyi.ai.domain.vo.WxChatVo;
import com.ruoyi.ai.domain.vo.WxDialogueVo;
import com.ruoyi.ai.domain.vo.WxModelConfigVo;
import com.ruoyi.ai.enums.MfgEnum;
import com.ruoyi.ai.factory.ModelChatTemplate;
import com.ruoyi.ai.service.IWxDialogueService;
import com.ruoyi.ai.ws.ChatWebSocketManager;
import com.ruoyi.common.exception.ServiceException;
import com.ruoyi.common.utils.StringUtils;
import com.ruoyi.common.utils.redis.RedisUtils;
import io.github.imfangs.dify.client.DifyChatClient;
import io.github.imfangs.dify.client.DifyClient;
import io.github.imfangs.dify.client.DifyClientFactory;
import io.github.imfangs.dify.client.callback.ChatStreamCallback;
import io.github.imfangs.dify.client.enums.ResponseMode;
import io.github.imfangs.dify.client.event.ErrorEvent;
import io.github.imfangs.dify.client.event.MessageEndEvent;
import io.github.imfangs.dify.client.event.MessageEvent;
import io.github.imfangs.dify.client.model.DifyConfig;
import io.github.imfangs.dify.client.model.chat.ChatMessage;
import io.github.lnyocly.ai4j.listener.SseListener;
import io.github.lnyocly.ai4j.service.IChatService;
import io.github.lnyocly.ai4j.service.PlatformType;
import io.github.lnyocly.ai4j.service.factor.AiService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.time.Duration;
import java.util.List;
import java.util.concurrent.Semaphore;
import java.util.function.Consumer;
import java.util.stream.Collectors;

/**
 * Ollama模型服务
 */
@Slf4j
@Service
public class DifyModelChatService extends ModelChatTemplate {
    @Autowired
    private AiService aiService;

    @Autowired
    private IWxDialogueService wxDialogueService;

    public DifyModelChatService() {
        super(MfgEnum.DIFY.getType());
    }

    @Resource
    private ChatWebSocketManager chatWebSocketManager;

    @Override
    public String chatStream(List<WxChatVo> chatVoList, Long userId) {
        try {
            log.info("发送的请求为{}", chatVoList);

            // 定义消息处理函数，用于发送websocket消息
            Consumer<String> messageHandler = message -> {
                chatWebSocketManager.sendOneMessage(userId, message);
            };

            // 使用公共方法执行流式聊天
            String result = doChatStream(chatVoList, messageHandler);

            //发送对话结束标记，前端好判断本次对话已结束，允许下一轮对话
            chatWebSocketManager.sendOneMessage(userId, OVER_FLAG);
            return result;
        } catch (Exception e) {
            log.error("调用Dify接口出现问题：{}", e.getMessage());
            throw new ServiceException("调用Dify接口出现问题");
        }
    }

    @Override
    public String chatStream(List<WxChatVo> chatVoList, Long userId, String prompt) {
        if (chatVoList.size() == 1 && StringUtils.isNotBlank(prompt)) {
            WxChatVo promptVo = new WxChatVo();
            promptVo.setRole(Role.SYSTEM.getValue());
            promptVo.setContent(prompt);
            chatVoList.add(0, promptVo);
        }
        return chatStream(chatVoList, userId);
    }

    @Override
    public String chatStreamOnce(String content) {
        try {
            List<WxChatVo> chatVoList = Lists.newArrayList();
            WxChatVo wxChatVo = new WxChatVo();
            wxChatVo.setRole(Role.USER.getValue());
            wxChatVo.setContent(content);
            chatVoList.add(wxChatVo);

            // 使用公共方法执行流式聊天，不发送websocket消息
            return doChatStream(chatVoList, null);
        } catch (Exception e) {
            log.error("调用Dify接口出现问题：{}", e.getMessage());
            return "调用大模型接口失败，请稍后尝试";
        }
    }

    @Override
    public String chatStreamOnce(String content, String prompt) {
        try {
            // 构建聊天内容列表
            List<WxChatVo> chatVoList = Lists.newArrayList();

            // 添加系统提示
            if (StringUtils.isNotBlank(prompt)) {
                WxChatVo promptVo = new WxChatVo();
                promptVo.setRole(Role.SYSTEM.getValue());
                promptVo.setContent(prompt);
                chatVoList.add(promptVo);
            }

            // 添加用户消息
            WxChatVo userVo = new WxChatVo();
            userVo.setRole(Role.USER.getValue());
            userVo.setContent(content);
            chatVoList.add(userVo);

            // 使用公共方法执行流式聊天，不发送websocket消息
            return doChatStream(chatVoList, null);
        } catch (Exception e) {
            log.error("调用Dify接口出现问题：{}", e.getMessage());
            return "调用大模型接口失败，请稍后尝试";
        }
    }

    /**
     * 执行流式聊天的公共方法
     * @param chatVoList 聊天消息列表
     * @param messageHandler 消息处理函数，如果为null则不发送websocket消息
     * @return 聊天结果
     * @throws Exception 异常
     */
    private String doChatStream(List<WxChatVo> chatVoList, Consumer<String> messageHandler) throws Exception {
        WxChatVo wxChatVo = chatVoList.get(chatVoList.size() - 1);
        // 获取会话ID
        String conversationId = getConversationId(wxChatVo.getDialogueId());

        // 同步信号量
        Semaphore semaphore = new Semaphore(0);
        // 构造基本请求参数
        ChatMessage message = ChatMessage.builder()
            .query(wxChatVo.getContent())
            .user(wxChatVo.getWxUserId().toString())
            .responseMode(ResponseMode.STREAMING)
            .conversationId(conversationId)
            .build();
        // 构造自定义监听器
        final StringBuilder resultBuilder = new StringBuilder();
        DifyChatClient difyClient = getDifyClient();
        difyClient.sendChatMessageStream(message, new ChatStreamCallback() {
            @Override
            public void onMessage(MessageEvent event) {
                System.out.println("收到消息片段: " + event.getAnswer());
                resultBuilder.append(event.getAnswer());
                // 如果有消息处理函数，则发送消息
                if (messageHandler != null) {
                    messageHandler.accept(event.getAnswer());
                }
            }

            @Override
            public void onMessageEnd(MessageEndEvent event) {
                String newConversationId = event.getConversationId();
                log.info("消息结束,会话ID: " + newConversationId);
                log.info("消息结束，完整消息ID: " + event.getMessageId());

                // 更新会话ID（如有必要）
                updateConversationId(wxChatVo.getDialogueId(), conversationId, newConversationId);

                semaphore.release();
            }

            @Override
            public void onError(ErrorEvent event) {
                semaphore.release();
                log.info("错误: " + event.getMessage());
            }

            @Override
            public void onException(Throwable throwable) {
                semaphore.release();
                log.info("异常: " + throwable.getMessage());
            }
        });
        semaphore.acquire();
        return resultBuilder.toString();
    }

    /**
     * 获取会话ID
     * 优先从缓存获取，缓存没有则从数据库获取
     *
     * @param dialogueId 对话ID
     * @return 会话ID
     */
    private String getConversationId(Long dialogueId) {
        // 优先从缓存获取conversationId
        String conversationId = RedisUtils.getCacheObject("dify:csId:" + dialogueId);
        // 缓存没有则从数据库获取
        if(StringUtils.isEmpty(conversationId)){
            WxDialogueVo wxDialogueVo = wxDialogueService.queryById(dialogueId);
            if(wxDialogueVo != null) {
                conversationId = wxDialogueVo.getThirdDialogueId();
                // 如果数据库中有值，则更新到缓存
                if(StringUtils.isNotEmpty(conversationId)) {
                    RedisUtils.setCacheObject("dify:csId:" + dialogueId, conversationId);
                }
            }
        }
        return conversationId;
    }

    /**
     * 更新会话ID
     * 只有在新会话ID不为空且与当前不同时才更新
     *
     * @param dialogueId 对话ID
     * @param oldConversationId 原会话ID
     * @param newConversationId 新会话ID
     */
    private void updateConversationId(Long dialogueId, String oldConversationId, String newConversationId) {
        // 只有在conversationId为空或者发生变化时才更新数据库和缓存
        if(StringUtils.isNotEmpty(newConversationId) && !newConversationId.equals(oldConversationId)){
            WxDialogueBo wxDialogueBo = new WxDialogueBo();
            wxDialogueBo.setThirdDialogueId(newConversationId);
            wxDialogueBo.setDialogueId(dialogueId);
            wxDialogueService.updateByBo(wxDialogueBo);
            //缓存时间限制为3天，防止缓存东西过多
            RedisUtils.setCacheObject("dify:csId:" + dialogueId, newConversationId, Duration.ofDays(3L));
            log.info("已更新会话ID: {}", newConversationId);
        }
    }

    private DifyChatClient getDifyClient() {
        WxModelConfigVo config = getConfig();
        // 创建自定义配置
        DifyConfig difyConfig = DifyConfig.builder()
            .baseUrl(config.getAppid())
            .apiKey(config.getApiKey())
            .connectTimeout(5000)  // 连接超时（毫秒）
            .readTimeout(60000)    // 读取超时（毫秒）
            .writeTimeout(30000)   // 写入超时（毫秒）
            .build();
        return DifyClientFactory.createChatClient(difyConfig);
    }
}
